package com.niit.mobileDevide.screenon;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class ScreenOnReducer extends Reducer<Text, ScreenOnBean, Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<ScreenOnBean> values, Context context) throws IOException, InterruptedException {
        List<ScreenOnBean> allUsers = new ArrayList<>();
        double totalScreenOnTime = 0.0;
        int count = 0;

        // 收集所有用户数据并计算总屏幕开启时间
        for (ScreenOnBean val : values) {
            allUsers.add(val);
            totalScreenOnTime += val.getScreenOnTime();
            count++;
        }

        if (count > 0) {
            // 计算平均屏幕开启时间
            double avgScreenOnTime = totalScreenOnTime / count;

            // 分类用户为大于和小于等于平均屏幕开启时间的两组
            List<ScreenOnBean> aboveAvgUsers = allUsers.stream()
                    .filter(user -> user.getScreenOnTime() > avgScreenOnTime)
                    .collect(Collectors.toList());

            List<ScreenOnBean> belowOrEqualAvgUsers = allUsers.stream()
                    .filter(user -> user.getScreenOnTime() <= avgScreenOnTime)
                    .collect(Collectors.toList());

            // 统计大于平均屏幕开启时间的用户总数，并输出这个数量以及平均屏幕开启时间
            int aboveAvgCount = aboveAvgUsers.size();
            context.write(new Text("AboveAverageUserCount"), new Text(String.format("%d, %.2f", aboveAvgCount, avgScreenOnTime)));

            // 统计小于或等于平均屏幕开启时间的用户总数，并输出这个数量以及平均屏幕开启时间
            int belowOrEqualAvgCount = belowOrEqualAvgUsers.size();
            context.write(new Text("BelowOrEqualAverageUserCount"), new Text(String.format("%d, %.2f", belowOrEqualAvgCount, avgScreenOnTime)));
        }
    }
}